package com.cellulam.framework.mq;

import com.cellulam.framework.core.spring.SmartEnvironment;
import com.cellulam.framework.mq.listener.*;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.annotation.Configuration;

/**
 *
 */
@Configuration
public class MqListenPostProcessor implements BeanPostProcessor {

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (bean instanceof MqListener) {
            this.addMqConsumer((MqListener) bean);
        }
        return bean;
    }

    private void addMqConsumer(MqListener subscriber) {
        MqConsumer consumer = subscriber.getClass().getAnnotation(MqConsumer.class);

        if (consumer == null) {
            return;
        }
        String groupId = SmartEnvironment.env.resolveIfNeed(consumer.groupId());

        MqConsumerHolder.holder.addMqConsumer(MqConsumeInfo.builder()
                .groupId(groupId)
                .consumerConfig(RocketMqConsumerConfig.builder()
                        .groupId(groupId)
                        .concurrentConsumers(consumer.concurrentConsumers())
                        .topic(SmartEnvironment.env.resolveIfNeed(consumer.topic()))
                        .tags(SmartEnvironment.env.resolveIfNeed(consumer.tags()))
                        .suspendTimeMillis(consumer.suspendTimeMillis())
                        .maxReConsumeTimes(consumer.maxReConsumeTimes())
                        .consumeTimeout(consumer.consumeTimeout())
                        .build())
                .subscriber(subscriber)
                .build());
    }

}
